package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"github.com/gorilla/websocket"
	"github.com/satori/go.uuid"
	"net/url"
	"os"
	"os/signal"
	"sync"
	"talent.com/agency_im/common/msg"
	"talent.com/server/durotar.git/leaf"
	"talent.com/server/durotar.git/proto"
	"talent.com/server/durotar.git/tools/sample"
	"talent.com/server/durotar.git/tools/tz"
	"time"
)

var (
	allConn = &sync.Map{}
)

type Conn struct {
	sync.Mutex
	ID       uint64
	RealConn *websocket.Conn
}

const (
	toUserId = 35 // IM平台用户id
	token      = "9149403c-5a9a-11e9-9d01-88e9fe4e21cb"
	//token = "sfjgs1nd93wf9knsdgsldf"
	//token = "0ccef3e2-70bf-11e9-87e3-0a08fd60f7ac"
	serverHost = "127.0.0.1"
	//serverHost = "0.0.0.0"
	//serverHost = "3.1.171.240"
	//serverHost = "chat.im.cnyda.net"
	wsAddr     = serverHost + ":3655"
	userCount  = 1000
	benchmark  = false
)

func logout(conn interface{}) {
	//c, _ := allConn.Load(conn)
	//x, _ := c.(*Conn)
	fmt.Println("logout...")
}

func wsMsgHandler(closeChan chan struct{}, wg *sync.WaitGroup, conn *websocket.Conn) {
	wg.Add(1)

	sendMsg(&msg.AuthReq{
		Token: token,
	}, conn)

	for {
		select {
		case <-closeChan:
			fmt.Println("closing websocket")
			logout(conn)
			wg.Done()
			return
		default:
			_, data, err := conn.ReadMessage()
			if err != nil {
				fmt.Printf("!!!ws error: %v\n", err)
				wg.Done()
				return
			} else {
				handlePush(closeChan, data, conn)
			}
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func ping(close chan struct{}, conn *websocket.Conn) {
	for {
		select {
		case <-close:
			fmt.Println("Exit Pin ....")
			return
		default:
			sendMsg(&proto.Ping{
				Timestamp: tz.GetNowTsMs(),
			}, conn)
		}
		time.Sleep(5 * time.Second)
	}
}

// 模拟一个会话场景
func beginTopic(close chan struct{}, ctx *Conn) {
	sendMsg(&msg.InTopicReq{
		Topic: &msg.Topic{TopicType: msg.TOPIC_TYPE_PRIVATE, ID: toUserId},
	}, ctx.RealConn)

	var (
		bytes []byte
		err   error
	)

	// 选择支付分类
	//if bytes, err = json.Marshal(map[string]interface {
	//}{
	//	"type": "pay:select_pay_type",
	//	"data": map[string]interface{}{
	//		"payType": 5,
	//	},
	//}); err != nil {
	//	panic(err)
	//}

	// 选择金额
	if bytes, err = json.Marshal(map[string]interface {
	}{
		"type": "pay:select_pay_amount",
		"data": map[string]interface{}{
			"receiptAddressId": 6,
			"amount":           100,
		},
	}); err != nil {
		panic(err)
	}

	sendMsg(&msg.ChatReq{
		ClientID: fmt.Sprintf("%v", uuid.NewV1()),
		MsgType:  msg.MSG_TYPE_CUSTOM,
		Topic:    &msg.Topic{TopicType: msg.TOPIC_TYPE_PRIVATE, ID: toUserId},
		Info:     string(bytes),
	}, ctx.RealConn)

//	go func() {
//		for {
//			select {
//			case <-close:
//				fmt.Println("Exit topic ....")
//				return
//			case <-time.Tick(time.Duration(rand.Intn(10)+3) * time.Second):
//				fmt.Printf("Send msg From %v to %v\n", ctx.ID, toUserId)
//				sendMsg(&msg.ChatReq{
//					ClientID: fmt.Sprintf("%v", uuid.NewV1()),
//					MsgType:  msg.MSG_TYPE_TEXT,
//					Topic:    &msg.Topic{TopicType: msg.TOPIC_TYPE_PRIVATE, ID: toUserId},
//					Info:     fmt.Sprintf("Send Text From %v to %v", ctx.ID, toUserId),
//				}, ctx.RealConn)
//			}
//		}
//	}()
}

func handlePush(close chan struct{}, data []byte, conn *websocket.Conn) {

	m, err := leaf.MsgProcessor.Unmarshal(data)
	if err != nil {
		fmt.Println(err)
	} else {
		switch m.(type) {
		case *msg.AuthResp:
			x := m.(*msg.AuthResp)
			ctx := &Conn{
				ID:       x.UserID,
				RealConn: conn,
			}
			allConn.Store(conn, ctx)
			fmt.Printf("<======== auth resp: %v\n", x)
			go ping(close, conn)
			go beginTopic(close, ctx)
		case *msg.ChatReqAck:
			x := m.(*msg.ChatReqAck)
			fmt.Printf("<======== chat req ack: %v\n", x)
		case *msg.ChatNT:
			x := m.(*msg.ChatNT)
			fmt.Printf("<======== chat NT: %v\n", x)
			sendMsg(&msg.ChatNTAck{
				Topic:    x.Topic,
				MaxMsgID: x.Msgs[x.MsgCount-1].MsgID},
				conn)
		case *proto.ErrorST:
			x := m.(*proto.ErrorST)
			fmt.Printf("error happened!!!, status:%v, msg:%s\n", x.Status, x.Msg)
		case *proto.Pong:
			fmt.Println("========================pong===================")
		default:
			if !benchmark {
				fmt.Printf("<========== received %+v\n", m)
			}
		}
	}
}

var addr = flag.String("addr", wsAddr, "http service address")

func getWSconn() *websocket.Conn {
	u := url.URL{Scheme: "ws", Host: *addr, Path: "/ws"}
	var dialer *websocket.Dialer
	wsconn, _, err := dialer.Dial(u.String(), nil)
	if err != nil {
		panic(err)
	}
	return wsconn
}

func initWSConn() *websocket.Conn {
	wsconn := getWSconn()
	return wsconn
}

func initWSConns(wg *sync.WaitGroup, closeChan chan struct{}) {
	for i := uint64(1); i <= userCount; i++ {
		time.Sleep(time.Duration(sample.RandInt(20, 100)) * time.Millisecond)
		wsconn := getWSconn()
		go wsMsgHandler(closeChan, wg, wsconn)
		fmt.Printf("User%v connected\n", i)
	}
}

func sendMsg(msg interface{}, conn *websocket.Conn) {
	if cx, ok := allConn.Load(conn); ok {
		cxt := cx.(*Conn)
		cxt.Lock()
		defer cxt.Unlock()
	}
	sendWSProtoMsg(msg, conn)
}

func sendWSProtoMsg(msg interface{}, wsconn *websocket.Conn) {
	data, err := leaf.MsgProcessor.Marshal(msg)
	if err != nil {
		fmt.Println("======> marshal error", err)
		panic(err)
	}
	// id + data
	m := make([]byte, 2+len(data[1]))
	// 默认使用大端序
	copy(m[0:], data[0])
	copy(m[2:], data[1])
	// 发送消息
	if err := wsconn.WriteMessage(websocket.BinaryMessage, m); err != nil {
		fmt.Println("write msg : ", err)
		panic(err)
	}
}

func main() {
	wg := &sync.WaitGroup{}
	closeChan := make(chan struct{})
	if benchmark {
		initWSConns(wg, closeChan)
	} else {
		wsconn := initWSConn()
		defer wsconn.Close()
		go wsMsgHandler(closeChan, wg, wsconn)
	}

	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt, os.Kill)
	<-c
	fmt.Println("logout")
	close(closeChan)
	wg.Wait()
	fmt.Println("all done")
}
